/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nageoffer.shortlink.project.config;

import com.nageoffer.shortlink.project.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;

/**
 * Redis Stream 消息队列配置
 * 公众号：马丁玩编程，回复：加群，添加马哥微信（备注：link）获取项目资料
 */
// Redis Stream消息队列配置类
// 负责配置短链接统计数据的消息队列消费者端
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    // Redis连接工厂，用于创建与Redis服务器的连接
    private final RedisConnectionFactory redisConnectionFactory;

    // 统计消息的消费者实现类
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;


    // 创建异步消费线程池的Bean
    @Bean
    public ExecutorService asyncStreamConsumer() {
        // 线程编号计数器
        AtomicInteger index = new AtomicInteger();
        // 获取CPU核心数作为基础线程数
        int processors = Runtime.getRuntime().availableProcessors();

        // 创建线程池配置：
        // - 核心线程数 = CPU核心数
        // - 最大线程数 = 核心数 + 核心数/2（即1.5倍）
        // - 空闲线程存活时间60秒
        // - 使用无界队列（LinkedBlockingQueue）
        // - 自定义线程命名规则（stream_consumer_short-link_stats_+序号）
        // - 设置为守护线程（daemon）
        return new ThreadPoolExecutor(processors,
                processors + processors >> 1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                runnable -> {
                    Thread thread = new Thread(runnable);
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }

    // 创建Stream消息监听容器的Bean
    // initMethod/destroyMethod指定生命周期方法
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        // 配置监听容器选项：
        // - 每次批量拉取10条消息
        // - 使用自定义的异步消费线程池
        // - 设置拉取超时时间为3秒（需小于Redis超时配置）
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .batchSize(10)
                        .executor(asyncStreamConsumer)
                        .pollTimeout(Duration.ofSeconds(3))
                        .build();

        // 创建监听容器实例（传入Redis连接工厂和配置选项）
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 配置消费者监听：
        // - 使用指定的消费者组（group参数）和消费者名称（"stats-consumer"）
        // - 从最后消费的位置开始读取（ReadOffset.lastConsumed()）
        // - 指定消息处理器（shortLinkStatsSaveConsumer）
        // - 自动确认模式（receiveAutoAck）
        streamMessageListenerContainer.receiveAutoAck(Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"),
                StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()), shortLinkStatsSaveConsumer);

        return streamMessageListenerContainer;
    }
}